#include "../include/SockConnector.h"
#include "../include/EpollProactor.h"
#include "../include/Proactor.h"
#include "../include/AsynchIO.h"
#include "../include/TimeValue.h"
#include "../include/NDK.h"

#include "../logger/Logger.h"

#include <cassert>
#include <deque>

enum
{
    MSG_LOGIN           = 200,
    MSG_LOGIN_RESP      = 201,
    MSG_HEART_BEAT      = 202
};
struct MsgHeader
{
    int msgid;
    int msglen;  // not include header
    MsgHeader ()
    : msgid (-1)
    , msglen (0)
    {}
};
struct LoginInfo
{
    MsgHeader header;
    //
    int  area_id;
    int  service_id;
    char username[16];
    char userpasswd[16];
    char reserve[8];

    LoginInfo ()
    : area_id (0)
    , service_id (0)
    {
        memset (username, 0, sizeof (username));
        memset (userpasswd, 0, sizeof (userpasswd));
        memset (reserve, 0, sizeof (reserve));
    }
};
struct LoginReturnInfo
{
    MsgHeader header;
    //
    int  result;
    int  userid;
    char reserve[8];
};
struct HeartBeatInfo
{
    MsgHeader header;
    int userid;
};
THREAD_FUNC_RETURN_T event_loop (void *arg)
{
    Proactor *p = reinterpret_cast<Proactor *> (arg);
    p->run_proactor_event_loop ();
    return 0;
}
#define INIT_BUFF_LEN          4096 
static Logger *client_log = LogManager::instance ()->get_logger ("client");
class Client : public AsynchHandler
{
public:
    Client () 
    : write_stream_buff_ (0)
    {
        bytes_transfer_ = 0;
        closed_ = 1;
        recv_resp_ = 0;
        write_count_ = 0;
        login_succ_ = 0;
        stop_read_ = 0;
        stop_write_ = 0;
        userid_ = 0;
    }
    ~Client ()
    { 
    }
    virtual int open (NDK_HANDLE handle)
    {
        this->read_stream_buff_ = new MessageBlock (INIT_BUFF_LEN);
        if (this->read_stream_.open (this, handle) == 0)
        {
            if (this->start_read () != 0)
            {
                client_log->error ("start read faild");
                return -1;
            }
        }else
            client_log->error ("read stream open faild");

        if (this->write_stream_.open (this, handle) != 0)
        {
            client_log->error ("write stream open faild");
            return -1;
        }
        this->closed_ = 0;
        return 0;
    }
    int closed ()
    {
        return this->closed_;
    }
    int login_succ ()
    {
        return this->login_succ_;
    }
    int recv_resp ()
    {
        return this->recv_resp_;
    }
    int userid ()
    {
        return this->userid_;
    }
    int start_read ()
    {
        return this->read_stream_.read (this->read_stream_buff_, 
                this->read_stream_buff_->size ()); 
    }
    int start_write (MessageBlock *mb)
    {
        assert (mb);
        int result = this->write_stream_.write (mb, mb->size ());
        if (result == 0) ++this->write_count_;
        return result;
    }
    virtual int handle_close (const int mask)
    {
        if (mask == AsynchHandler::READ_MASK)
        {
            NDK_OS::closesocket (this->handle ());
            //client_log->rinfo ("recv data length = %d", this->bytes_read_);
            if (this->read_stream_buff_ == 0)
                delete this->read_stream_buff_;
            this->read_stream_buff_ = 0;
            this->stop_read_ = 1;
        }else if (mask == AsynchHandler::WRITE_MASK)
        {
            //client_log->rinfo ("send data length = %d", this->bytes_write_);
            if (this->write_stream_buff_ == 0)
                delete this->write_stream_buff_;
            this->write_stream_buff_ = 0;
            this->stop_write_ = 1;
        }
        if (this->stop_read_ && this->stop_write_)
        {
            client_log->debug ("release handler");
            delete this;
        }
        return 0;
    }
    virtual int handle_write_stream (const AsynchWriteStream::Result &result)
    {
        --this->write_count_;
        if (!result.success ())
        {
            client_log->error ("writestream not success, errno = %d",
                    result.error ());
            if (result.error () != EAGAIN)
            {
                return -1;
            }
        }else if (result.bytes_transferred () == 0)
        {
           client_log->error ("peer closed");
            return -1;
        }

        result.message_block ()->release ();
        if (this->write_count_ == 0)
            this->stop_write_ = 1;
        else
            this->stop_write_ = 0;
        return 0;
    }
    virtual int handle_read_stream (const AsynchReadStream::Result &result)
    {
        if (!result.success ())
        {
            client_log->error ("read stream not success, errno = %d",
                    result.error ());
            if (result.error () != EAGAIN)
            {
                return -1;
            }
        }else if (result.bytes_transferred () == 0)
        {
            client_log->error ("server: closed");
            return -1;
        }
        // read header.
        MessageBlock *message_block = result.message_block ();
        MsgHeader *header = 
            reinterpret_cast<MsgHeader *>(message_block->base ());
        if (header)
        {
            if (header->msgid == MSG_LOGIN_RESP)
            {
                this->recv_resp_ = 1;
                LoginReturnInfo *retinfo = 
                    reinterpret_cast<LoginReturnInfo*>(message_block->base ());
                if (retinfo->result == 0)
                {
                    this->userid_ = retinfo->userid;
                    this->login_succ_ = 1;
                }
            }
        }
        return 0;
    }
protected:
    AsynchReadStream read_stream_;

    AsynchWriteStream write_stream_;

    MessageBlock *write_stream_buff_;

    MessageBlock *read_stream_buff_;
    int bytes_transfer_;

    int closed_;

    int recv_resp_;

    int login_succ_;

    int write_count_;

    int stop_read_;

    int stop_write_;

    int userid_;
};
static Logger *main_log = LogManager::instance ()->get_logger ("root");
int main (int argc, char *argv[])
{
    signal (SIGPIPE, SIG_IGN);
    const char *usage = "./prog 192.168.1.1 8889 clientnum interval(msec) packnum";
    if (argc != 6) 
    {
        printf ("%s\n", usage);
        return -1;
    }
    EpollProactor *ep = new EpollProactor ();
    if (ep->open () != 0)
    {
        fprintf (stderr, "epoll proactor open failed\n");
        return -1;
    }
    LogManager::instance ()->init ("clog.ini");
    Proactor *p = new Proactor (ep, 1);
    Proactor::instance (p);
    ThreadManager::instance ()->spawn_n (1, event_loop, Proactor::instance ());
    int clnum = atoi(argv[3]);
    std::deque<Client *> client_list;
    for (int i = 0; i < clnum; ++i)
    {
        SockStream peer;
        SockConnector connector;
        InetAddr addr (atoi(argv[2]), argv[1]);
        if (connector.connect (peer, addr) == 0)
        {
            Client *c = new Client;
            c->open (peer.handle ());
            client_list.push_back(c);
        }
        
    }
    printf ("connect success : %d, failed : %d\n", 
            client_list.size(),
            clnum - client_list.size());
    // 1. login
    std::deque<Client *>::iterator itor;
    int count = 0x9999;
    for (itor = client_list.begin (); itor != client_list.end ();)
    {
        MessageBlock *msg = new MessageBlock (sizeof (LoginInfo));
        LoginInfo *login = reinterpret_cast<LoginInfo*>(msg->base ());
        login->header.msgid  = MSG_LOGIN;
        login->header.msglen = sizeof (LoginInfo) - sizeof (MsgHeader);
        login->area_id = 10;
        login->service_id = 20;
        snprintf (login->username, 16, "id=%d", count++);
        strcpy (login->userpasswd, "passwd");
        strcpy (login->reserve, "911");
        msg->wr_ptr (sizeof (LoginInfo));
        if ((*itor)->start_write (msg) != 0)
        {
            main_log->trace ("msg size = %d", msg->size ());
            main_log->error ("start write error!!");
            itor = client_list.erase (itor);
            continue;
        }
        ++itor;
    }
    while (1)
    {
        sleep (2);
        for (itor = client_list.begin (); itor != client_list.end ();)
        {
            if ((*itor)->closed ())
            {
                itor = client_list.erase (itor);
                continue;
            }
            if ((*itor)->login_succ ())
            {
                MessageBlock *msg = new MessageBlock (sizeof (HeartBeatInfo));
                HeartBeatInfo *hbinfo = reinterpret_cast<HeartBeatInfo*>(msg->base ());
                hbinfo->header.msgid  = MSG_HEART_BEAT;
                hbinfo->header.msglen = sizeof (HeartBeatInfo) - sizeof (MsgHeader);
                hbinfo->userid = (*itor)->userid ();
                msg->wr_ptr (sizeof (HeartBeatInfo));
                (*itor)->start_write (msg);
            }else
            {
                // 
                if ((*itor)->recv_resp ())
                {
                    itor = client_list.erase (itor);
                    continue;
                }
            }
            itor++;
        }
    }
    ThreadManager::instance ()->wait ();
    return 0;
}
